package com.kafka.cn.consumer;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

/**
 * @author yangshen
 * @date 2020/3/31 23:02
 *  自定义提交offset
 */
public class CustomConsumer {

    private static Map<TopicPartition, Long> currentOffset = new HashMap<>();

    public static void main(String[] args) {
        //创建配置信息
        Properties properties = new Properties();
        //Kafka 集群
        properties.put("bootstrap.servers", "hadoop102:9092");
        //消费者组，只要 group.id 相同，就属于同一个消费者组
        properties.put("group.id", "test");
        //关闭自动提交 offset
        properties.put("enable.auto.commit", "false");
        //Key 和 Value 的反序列化类
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //创建一个消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        //消费者订阅主题
        consumer.subscribe(Arrays.asList("first", ""), new ConsumerRebalanceListener() {
            //该方法会在Rebalance(重新分配)之前调用
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                //提交该消费者所有分区的 offset
                commitOffset(currentOffset);
            }

            //该方法会在Rebalance(重新分配)之后调用
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                currentOffset.clear();
                for (TopicPartition partition : partitions) {
                    //定位到最近提交的 offset 位置继续消费,重新分配之前的offset是另外一个消费者(已挂)消费的，现在要接着上一个消费者的offset消费
                    consumer.seek(partition, getOffset(partition));
                }
            }
        });

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);//消费者拉取数据
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                currentOffset.put(new TopicPartition(record.topic(),
                        record.partition()), record.offset());
            }
            commitOffset(currentOffset);//异步提交
        }
    }
    //提交该消费者所有分区的 offset
    private static void commitOffset(Map<TopicPartition, Long> currentOffset){
        //自己实现
        //如果想保存到mysql，就保存到mysql中，做一个联合主键(4个字段：消费者组group,主题topic，分区partition，偏移量offset)gtpo
    }

    //获取某分区的最新 offset
    private static long getOffset(TopicPartition partition){
        //自己实现
        //同理 也要到mysql中读
        return 0;
    }

}
